package com.xy.springboot_3.source.demo;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * jdk 1.9 提供的 发布者，订阅者
 */
@Slf4j
public class SubmissionDemo {


    public static void main(String[] args) throws InterruptedException {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 一定要绑定订阅关系
        Flow.Subscriber subscriber = new Flow.Subscriber() {


            private Flow.Subscription subscription;

            /**
             * 订阅者每次发送消息触发
             * @param subscription a new subscription
             */

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription= subscription;
                log.info("onSubscribe");
                // 像订阅者再次请求数据
                subscription.request(1L);
            }

            /**
             * 订阅下个元素处理前触发
             * @param item the item
             */
            @Override
            public void onNext(Object item) {
                log.info("onNext"+item);
                subscription.request(1L);

            }

            /**
             * 订阅处理中产生异常触发
             * @param throwable the exception
             */
            @Override
            public void onError(Throwable throwable) {
                log.info("onError");
            }

            /**
             * 订阅结束完成后出发
             */
            @Override
            public void onComplete() {
          log.info("onComplete");
            }
        };
        // 绑定订阅者
        publisher.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            publisher.submit(String.valueOf(i));
        }



        // 发布者通道关闭
        publisher.close();

        TimeUnit.SECONDS.sleep(3L);
    }


}
